package mapred.distinctData.custom;


import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

import java.io.IOException;

public class DistinctJob {
    public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException {
        // 判断当前如数参数的个数是否有效
        if (args.length != 2) {
//            serr
            System.err.println("请输入2个参数");
            System.exit(2);
        }

        // 创建配置类的对象，此对象会自动从Hadoop的各个配置文件读取相关数据
        Configuration conf = new Configuration();
        // 通过单例模式创建Job对象的实例
        Job job = Job.getInstance(conf);

        // 设置Jar包的信息
        job.setJobName("Distinct Data");
        job.setJarByClass(DistinctJob.class);

        // 设置Mapper类和Reducer类
        job.setMapperClass(DistinctMapper.class);
        job.setReducerClass(DistinctReducer.class);

        // 单独设置Mapper的输出的Kev-Value的类型
        // 也可以不设置，但是，默认会与Reducer的输出的K-V类型一致
        job.setMapOutputKeyClass(Text.class);
        job.setMapOutputValueClass(NullWritable.class);

        // 设置Reducer的输出K-V的类型
        // 默认与Mapper的输出K-V类型一致，除非Mapper通过setMapOutputXXXClass单独设置
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(NullWritable.class);

        // 设置Job的输入文件夹和输出文件夹
        // 使用FileInputFormat和FileOutputFormat的静态方法来设置
        // 接受的参数类型是Path类型
        FileInputFormat.addInputPath(job, new Path(args[0]));
        FileOutputFormat.setOutputPath(job, new Path(args[1]));

        System.exit(job.waitForCompletion(true) ? 0 : 1);
    }
}
